package com.sunzm.spark.streaming.callcenter

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.{KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 版本1，实现通话个数（呼入数，呼入成功数、呼出数，呼出成功数）的统计
 * 问题：无法统计时长类的（比如坐席振铃时长，需要接听时间-振铃时间，收到接听消息的时候，没有保存振铃时间）
 */
object CallCenterStaffJobV1 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName.stripSuffix("$")}")
      .setMaster("local[*]")

    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val kafkaParams = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "82.156.210.70:9093",
      ConsumerConfig.GROUP_ID_CONFIG -> "group_id_call_center",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer])

    val topics = Array("call-center-message")
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //把消息中的value取出来（不需要key）
    val kafkaMessageDStream: DStream[String] = kafkaDStream.map(_.value)

    //开始进行处理
    processDStream(kafkaMessageDStream)

    //启动程序
    ssc.start()
    ssc.awaitTermination()
  }

  def processDStream(kafkaMessageDStream: DStream[String]) = {

    //每天每个公司每个坐席的通话数量
    val staffCallCountDS = kafkaMessageDStream.transform(rdd => {

      rdd.filter(line => {
        //先把格式异常的数据过滤掉
        var res = false

        try {
          val jSONObject = JSON.parseObject(line)

          if (jSONObject.containsKey("companyId") &&
            jSONObject.containsKey("staffId") &&
            jSONObject.containsKey("dateTime")) {
            res = true
          }else{
            println(s"缺少字段[companyId,staffId,dateTime]: ${line}")
          }
        } catch {
          case _ => {
            println(s"JSON格式异常: ${line}")
          }
        }

        res
      }).map(line => {
        //解析JSON
        val jSONObject = JSON.parseObject(line)

        //公司ID
        val companyId = jSONObject.getString("companyId")
        //坐席ID
        val staffId = jSONObject.getString("staffId")
        //时间 yyyyMMddHHmmssSSS -> yyyy-MM-dd
        val dateTimeStr = jSONObject.getString("dateTime")
        //对字符串的日期进行格式转换
        val dateTimeLong: Long = DateUtils.parseDate(dateTimeStr, "yyyyMMddHHmmssSSS").getTime
        //转换成 yyyy-MM-dd 格式
        val resultDate: String = DateFormatUtils.format(dateTimeLong, "yyyy-MM-dd")

        //开始统计
        //呼入方向： 0呼入 1呼出
        val direction: Int = jSONObject.getIntValue("direction")
        /**
         * 消息类型:
         *  start通话开始; staff-ring坐席振铃; staff-answer坐席接听;
         *  staff-hunup坐席挂机; user-ring：用户振铃； user-answer：用户接听;
         *  user-hunup：用户挂断; stop:通话结束
         */
        val messageType: String = jSONObject.getString("messageType")

        //呼入电话数量
        val callInCount = if(direction == 0 && StringUtils.equals("start", messageType)) 1 else 0
        //呼入成功数量
        var callInSuccCount = 0
        if(direction == 0 && StringUtils.equals("staff-answer", messageType)){
          //呼入，坐席接听，就表示呼入成功
          callInSuccCount = 1
        }
        //呼出电话数量
        val callOutCount = if(direction == 1 && StringUtils.equals("start", messageType)) 1 else 0
        //呼出成功数量
        var callOutSuccCount = 0
        if(direction == 1 && StringUtils.equals("user-answer", messageType)){
          //呼出，用户接听，就表示呼出成功
          callOutSuccCount = 1
        }

        ((resultDate, companyId, staffId), (callInCount, callInSuccCount, callOutCount, callOutSuccCount))
      }).reduceByKey((t1, t2) => {
        val callInCount: Int = t1._1 + t2._1
        val callInSuccCount: Int = t1._2 + t2._2
        val callOutCount = t1._3 + t2._3
        val callOutSuccCount = t1._4 + t2._4

        (callInCount, callInSuccCount, callOutCount, callOutSuccCount)
      })

    })


    staffCallCountDS.print()

  }
}
